package com.example;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 测试分布式锁
 */
public class DistributeLock {

    private static final int sessionTimeout = 2000;
    private static final String connectString = "Cent7:2181,Cent7-c1:2181,Cent7-c2:2181";
    private ZooKeeper zkServer = null;
    private CountDownLatch connectionLatch = new CountDownLatch(1);
    private CountDownLatch listenerLatch = new CountDownLatch(1);
    private String waitPath;
    private String currentNode;

    public DistributeLock() throws IOException, InterruptedException, KeeperException {

        // 获取连接
        zkServer = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

            @Override
            public void process(WatchedEvent watchedEvent) {
                // 如果连接上，就释放connectionLatch
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    connectionLatch.countDown();
                }

                // 释放listenerLatch
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
                    listenerLatch.countDown();
                }
            }
        });

        // 等待上一个线程
        connectionLatch.await();

        // 判断根节点/locks是否存在
        Stat stat = zkServer.exists("/locks", false);

        if (stat == null) {

            System.out.println("根节点/locks不存在，将自动创建");

            // 创建根节点
            zkServer.create("/locks", "锁的根节点".getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    /**
     * 加锁
     */
    public void lock() throws InterruptedException, KeeperException {

        // 在/locks下创建临时的，带序号的节点
        currentNode = zkServer.create("/locks/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        // 判断节点是否为最小序号，如果是，则获得锁，如果不是，则监听上一个节点
        List<String> children = zkServer.getChildren("/locks", false);

        if (children.size() == 1) {

            // 获得锁
            return;
        }

        // 对序号排序
        Collections.sort(children);

        // 获取节点的序号
        String thisNode = currentNode.substring("/locks/".length());

        // 获取当前节点在集合中的位置
        int index = children.indexOf(thisNode);

        // 判断当前节点
        if (index == -1) {
            throw new RuntimeException("children集合数据异常");
        }
        if (index == 0) {

            // 可以获取锁

        } else {
            waitPath = "/locks/" + children.get(index - 1);
            zkServer.getData(waitPath, true, null);

            // 等待监听完成
            listenerLatch.await();

        }

        // 最小序号完成业务后被删除然后释放锁

    }

    /**
     * 释放锁
     */
    public void unlock() throws InterruptedException, KeeperException {

        zkServer.delete(currentNode, -1);
    }

}
